package com.deep.flink;



import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @author wangshida@baijia.com
 * @datetime 2022-03-20 下午6:22
 * @CopyRight (C) 百家互联
 * @desc
 * @menu
 */
public class FLink04CustomSourceApp {

    public static void main(String[] args) throws Exception {
        //WebUi方式运行
        final StreamExecutionEnvironment env =
                StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置运行模式为流批一体
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //并行度
        env.setParallelism(2);
        //设置为自定义source
        DataStream<VideoOrder> ds = env.addSource(new VideoOrderSource());

        //过滤
        DataStream<VideoOrder> dfFliter = ds.filter(new FilterFunction<VideoOrder>() {
            @Override
            public boolean filter(VideoOrder videoOrder) throws Exception {
                return videoOrder.getMoney() > 10;
            }
        }).setParallelism(3);

        dfFliter.print().setParallelism(4);

        //设置名字
        env.execute("CustomSourceApp");
    }
}